package com.bw.gmall.reatime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.reatime.dwd.db.app
 * @className: DwdTradeRefundPaySucDetail
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/2/27 11:02
 * @version: 1.0
 */

public class DwdTradeRefundPaySucDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeRefundPaySucDetail().start(Constant.TOPIC_DB, Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS, 4, 10018);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        //读取topic_db
        readOdsDb(tableEnv, groupId);
        //读取dic
        readBaseDic(tableEnv);
        // 3. 过滤退款成功表数据
        Table getRefundPayment = getGetRefundPayment(tableEnv);
        tableEnv.createTemporaryView("refund_payment",getRefundPayment);
//        getRefundPayment.execute().print();
        //过滤退单表中的退单成功的数据
        Table getOrderRefundInfo = getGetOrderRefundInfo(tableEnv);
        tableEnv.createTemporaryView("order_refund_info",getOrderRefundInfo);
//        getRefundPayment.execute().print();

        //获取订单表中的退款成功的数据
        Table getOrderInfo = getGetOrderInfo(tableEnv);
        tableEnv.createTemporaryView("order_info",getOrderInfo);
//        getOrderInfo.execute().print();
        //关联
        Table joinTable = getJoinTable(tableEnv);
//        joinTable.execute().print();

        //写入kakfa
        extracted(tableEnv);
        joinTable.executeInsert(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS);
    }

    private static void extracted(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS+"(\n" +
                "    id string,\n" +
                "    user_id string,\n" +
                "    province_i string,\n" +
                "    order_id string,\n" +
                "    sku_id string,\n" +
                "    payment_type string,\n" +
                "    payment_type_name string,\n" +
                "    date_id string,\n" +
                "    callback_time string,\n" +
                "    refund_type  string,\n" +
                "    refund_num  string,\n" +
                "    refund_amount  string,\n" +
                "    total_amount string,\n" +
                "    ts bigint\n" +
                ")"+ SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS));
    }


    private static Table getJoinTable(StreamTableEnvironment tableEnv) {
        return getJoinTable(tableEnv, "select\n" +
                "    rp.id,\n" +
                "    oi.user_id,\n" +
                "    oi.province_id,\n" +
                "    rp.order_id,\n" +
                "    rp.sku_id,\n" +
                "    rp.payment_type,\n" +
                "    c.info.dic_name payment_type_name,\n" +
                "    date_format(rp.callback_time,'yyyy-MM-dd') date_id,\n" +
                "    rp.callback_time,\n" +
                "    ori.refund_type ,\n" +
                "    ori.refund_num ,\n" +
                "    ori.refund_amount ,\n" +
                "    rp.total_amount,\n" +
                "    rp.ts\n" +
                "from refund_payment rp\n" +
                "join order_refund_info ori on rp.order_id = ori.order_id and rp.sku_id = ori.sku_id\n" +
                "join order_info oi on rp.order_id = oi.id\n" +
                "join base_dic FOR SYSTEM_TIME AS OF rp.proc_time AS c ON rp.payment_type = c.rowkey;");
    }

    private static Table getJoinTable(StreamTableEnvironment tableEnv, String s) {
        return tableEnv.sqlQuery(s);
    }

    private static Table getGetOrderInfo(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['province_id'] province_id\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "  and `table` = 'order_info'\n" +
                "  and `type` = 'update'\n" +
                "  and `old`['order_status'] is not null\n" +
                "  and `data`['order_status'] ='1006'");
    }

    private static Table getGetOrderRefundInfo(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id ,\n" +
                "    `data`['order_id'] order_id ,\n" +
                "    `data`['sku_id'] sku_id ,\n" +
                "    `data`['refund_type'] refund_type ,\n" +
                "    `data`['refund_num'] refund_num ,\n" +
                "    `data`['refund_amount'] refund_amount ,\n" +
                "    `data`['refund_num'] refund_num\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "  and `table` = 'order_refund_info'\n" +
                "  and `type` = 'update'\n" +
                "  and `old`['refund_status'] is not null\n" +
                "  and `data`['refund_status'] ='0705'");
    }

    private static Table getGetRefundPayment(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['order_id'] order_id,\n" +
                "    `data`['sku_id'] sku_id,\n" +
                "    `data`['payment_type'] payment_type,\n" +
                "    `data`['total_amount'] total_amount,\n" +
                "    `data`['callback_time'] callback_time,\n" +
                "    ts,\n" +
                "    proc_time\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table` = 'refund_payment'\n" +
                "and `type` = 'update'\n" +
                "and `old`['refund_status'] is not null\n" +
                "and `data`['refund_status'] ='1602'");
    }
}
